package org.spdata.tsdb.service.impl;

import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import javax.annotation.PostConstruct;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.spdata.tsdb.MemTable;
import org.spdata.tsdb.client.DownSample;
import org.spdata.tsdb.client.DownSample.Arithmetic;
import org.spdata.tsdb.client.Query;
import org.spdata.tsdb.client.SubQuery;
import org.spdata.tsdb.dao.DpsDAO;
import org.spdata.tsdb.dao.MetricDAO;
import org.spdata.tsdb.dto.DataPoint;
import org.spdata.tsdb.dto.DataPoints;
import org.spdata.tsdb.entity.DpsPack;
import org.spdata.tsdb.entity.Metric;
import org.spdata.tsdb.entity.Summary;
import org.spdata.tsdb.service.DataPointService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.mongodb.client.model.Filters;

@Service
public class DataPointServiceImpl implements DataPointService {
	private static Logger logger = LoggerFactory.getLogger(DataPointServiceImpl.class);
	
	private Map<String,Metric> metrics = new ConcurrentHashMap<String,Metric>();
	
	@Autowired
	MetricDAO metricDAO = null;
	
	@Autowired
	DpsDAO dpsDAO = null;
	
	@Autowired
	MemTable memTable = null;
	
	
	private volatile AtomicBoolean writeLock = new AtomicBoolean(false);
	
		
	public DataPointServiceImpl(){
		
	}
	
	public Metric getMetric(int id) {
		Metric metric = metricDAO.findOne(Filters.eq("_id",id));
		return metric;
	}

	public MetricDAO getMetricDAO() {
		return metricDAO;
	}

	public void setMetricDAO(MetricDAO metricDAO) {
		this.metricDAO = metricDAO;
	}

	
	public DpsDAO getDpsInHourDAO() {
		return dpsDAO;
	}

	public void setDpsInHourDAO(DpsDAO dpsDAO) {
		this.dpsDAO = dpsDAO;
	}

	/**
	 * 初始化
	 */
	@PostConstruct
	public void init(){	
		List<Metric> metricsInDb = metricDAO.listAll();
		for(Metric metric :metricsInDb){
			metrics.put(metric.getKey(), metric);
		}
	}
	
	public Map<String, Metric> getMetrics() {
		return metrics;
	}

	public void setMetrics(Map<String, Metric> metrics) {
		this.metrics = metrics;
	}

	public void cleanMetrics(){
		this.metrics.clear();
	}
	
	/**
	 * 写日志时可能需要这个方法
	 * @param hour
	 * @return
	 */
	private String convertHourToString(long hour){		
		DateFormat df = new SimpleDateFormat("yyyy-MM-dd HH");
		return df.format(new Date(hour*3600*1000));
	}
	
	/**
	 * 批量处理采样数据
	 */
	public void dealDpList(List<DataPoint> dps) {
		boolean success = false;
		for(;;) {
			success=writeLock.compareAndSet(false, true);
			if(success) {
				logger.debug("获得写锁，写入时序数据,size={}",dps.size());
				long now = System.currentTimeMillis();
				for(DataPoint dp :dps){
					this.innerDealRealTimeDataPoint(dp, now);
				}
				writeLock.set(false);
				break;
			}else {
				try {
					Thread.sleep(1);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}		
	}

	/**
	 * 为采样数据补充测点信息，并缓存到memTable中
	 * @param dp
	 * @param time
	 */
	protected void innerDealRealTimeDataPoint(DataPoint dp,long time) {	
		Metric metric = this.findAndCreateMetric(dp);
		dp.setMetricId(metric.getId());
		if(dp.getTime()==null)
			dp.setTime(time);
		memTable.putDataPoint(dp);
	}
	
	private Metric getMetricFromDP(DataPoint dp) {
		Metric m = new Metric();
		m.setName(dp.getMetric());
		m.setTags(dp.getTags());
		return m;
	}
	

	private Metric findAndCreateMetric(DataPoint dp){
		Metric dto = this.getMetricFromDP(dp);
		String key = dto.getKey();		
		Metric metric = this.metrics.get(key);	
		if(metric==null){
			try {
				metric = metricDAO.createMetric(dto,dp.getValue());
			} catch (Exception e) {
				e.printStackTrace();
			}
			this.metrics.put(key, metric);
		}
		return metric;
	}
	
	public List<DataPoints> query(Query query) {
		List<DataPoints> result = new ArrayList<DataPoints>();
		Date start = query.getStart();
		Date end = query.getEnd();
		if(start==null)
			throw new RuntimeException("开始时间不能为空！");
		
		Long startSecond = start.getTime() / 1000;
		Long endSecond = null;
		if(end!=null)
			endSecond = end.getTime() / 1000;
		
		List<SubQuery> subQueries = query.getSubQueries();
		for(SubQuery q:subQueries){
			List<DataPoints> dpsList = this.doSubQuery(q,startSecond, endSecond,query.getSample());
			if(dpsList!=null)
				result.addAll(dpsList);
		}
		return result;
	}
	
	/**
	 * 清除超过时间范围的采样数据
	 * @param set
	 * @param startSecond
	 * @param endSecond
	 */
	private void removeDpsOverRange(List<DpsPack> list,Long startSecond,Long endSecond) {
		if(list!=null && list.size()>0){			
			list.get(0).subDps(startSecond, endSecond);
			if(list.size()>1){
				list.get(list.size()-1).subDps(startSecond, endSecond);
			}
		}
	}
	
	private void merge(Map<Metric,List<DpsPack>> target ,Map<Metric,List<DpsPack>> from){
		Set<Entry<Metric,List<DpsPack>>> entries = from.entrySet();
		for(Entry<Metric,List<DpsPack>> entry:entries) {
			Metric metric = entry.getKey();
			List<DpsPack> set = target.get(metric);
			if(set==null) {
				set = new ArrayList<DpsPack>();
				target.put(metric, set);
			}
			List<DpsPack> dpsList = entry.getValue();
			if(dpsList!=null)
				set.addAll(dpsList);
		}
	}
	
	private List<DataPoints> doSubQuery(SubQuery query,Long startSecond,Long endSecond,DownSample sample){	
		List<DataPoints> result = new ArrayList<DataPoints>();
		//从磁盘查询
		Map<Metric,List<DpsPack>> dpsFromDisk = this.dpsDAO.findDPS(query, startSecond, endSecond,sample);
		if(dpsFromDisk==null)
			dpsFromDisk = new HashMap<Metric,List<DpsPack>>();
		//从内存查询
		Map<Metric,List<DpsPack>> dpsFromMem = this.findDpsFromMem(query, startSecond, endSecond,sample);
		
		//合并
		Map<Metric,List<DpsPack>> merged = new LinkedHashMap<Metric,List<DpsPack>>();
		this.merge(merged, dpsFromDisk);
		this.merge(merged, dpsFromMem);
		
		Set<Entry<Metric,List<DpsPack>>> entries = merged.entrySet();
		for(Entry<Metric,List<DpsPack>> entry:entries) {
			Metric metric = entry.getKey();
			List<DpsPack> dpsSet = entry.getValue();
			this.removeDpsOverRange(dpsSet, startSecond, endSecond);
			DataPoints dps = this.downSample(dpsSet, sample);
			dps.setStart(startSecond*1000);
			dps.setEnd(endSecond*1000);
			dps.setMetric(metric.getName());
			dps.setTags(metric.getTags());
			result.add(dps);
		}
		return result;
	}
	
	
	private Map<Metric,List<DpsPack>> findDpsFromMem(
			SubQuery query,long startSecond,long endSecond,DownSample sample){		
		Map<Metric,List<DpsPack>> map = new HashMap<Metric,List<DpsPack>>();
		List<Metric> metrics = metricDAO.findMetrics(query.getMetric(), query.getTags());
		for(Metric metric:metrics) {
			List<DpsPack> list = new ArrayList<DpsPack>();
			DpsPack dpsOfCurrentHour = memTable.getDPS(metric.getId(),sample);
			DpsPack dpsOfNextHour = memTable.getDPSFromExtra(metric.getId(),sample);
			if(dpsOfCurrentHour!=null)
				list.add(dpsOfCurrentHour);
			if(dpsOfNextHour!=null)
				list.add(dpsOfNextHour);
			map.put(metric, list);
		}
		return map;
	}
	
	
	/**
	 * 数据合并 并 降频采样
	 */
	private DataPoints downSample(List<DpsPack> dpsInHours,DownSample sample){
		DataPoints dps = new DataPoints();
		if(sample==null){
			for(DpsPack dpsInHour:dpsInHours){
				long hour = dpsInHour.getHour();
				long zeroSecondOfHour = hour* 3600;
				Iterator<Entry<Integer,Number>> iter = dpsInHour.getDps().entrySet().iterator();
				while(iter.hasNext()){
					Entry<Integer,Number> entry = iter.next();
					long second = zeroSecondOfHour + entry.getKey();
					dps.addDataPoint(second*1000, entry.getValue());
				}
			}
		}else{
			TimeUnit unit = sample.getUnit();
			if(unit.equals(TimeUnit.MICROSECONDS) 
					|| unit.equals(TimeUnit.MILLISECONDS)
					|| unit.equals(TimeUnit.NANOSECONDS)
					|| unit.equals(TimeUnit.SECONDS))
				throw new RuntimeException("can not support downsample on this time unit:"+unit.toString());
			
			
			if(unit.equals(TimeUnit.HOURS) || unit.equals(TimeUnit.DAYS)){
				//从小时摘要数据降频采样
				Map<String,Number> sampled = sampleFromSummary(dpsInHours,sample);
				dps.setDps(sampled);
			}else{
				//从小时详细数据降频采样
				Map<String,Number> sampled = sampleFromDetail(dpsInHours,sample);
				dps.setDps(sampled);
			}
		}	
		return dps;
	}
	
	private int changeTimeLenToSeconds(int len,TimeUnit unit ){
		if(unit.equals(TimeUnit.DAYS)){
			return len * 24 * 3600;
		}else if(unit.equals(TimeUnit.HOURS)){
			return len * 3600;
		}else if(unit.equals(TimeUnit.MINUTES)){
			return len * 60;
		}else
			throw new RuntimeException("can not change time len to second with unit is "+ unit.toString());
	}
	
	private int changeTimeLenToHours(int len,TimeUnit unit ){
		if(unit.equals(TimeUnit.DAYS)){
			return len * 24 ;
		}else if(unit.equals(TimeUnit.HOURS)){
			return len ;
		}else
			throw new RuntimeException("can not change time len to second with unit is "+ unit.toString());
	}
	
	private Number sampleFromSummary(Summary summary,DownSample sample){
		Arithmetic arithmetic = sample.getArithmetic();
		if(arithmetic.equals(Arithmetic.avg))
			return summary.getAvg();
		else if(arithmetic.equals(Arithmetic.min))
			return summary.getMin();
		else if(arithmetic.equals(Arithmetic.max))
			return summary.getMax();
		else if(arithmetic.equals(Arithmetic.first))
			return summary.getBegin();
		else if(arithmetic.equals(Arithmetic.last))
			return summary.getEnd();
		else
			throw new RuntimeException("can not to execute sample Arithmetic:"+sample.getArithmetic().toString());
	}
	
	private Map<String,Number> sampleFromSummary(List<DpsPack> dpsInHours,DownSample sample){
		int frequenceLen = changeTimeLenToHours(sample.getFrequence(),sample.getUnit());
		long sampleStartAt = 0;
		Map<String,Number> result = new LinkedHashMap<String,Number>();
		Map<Long,Number> subList = new LinkedHashMap<Long,Number>();
		
		for(DpsPack dpsInHour:dpsInHours){			
			long hour = dpsInHour.getHour();	
			if(sampleStartAt==0){
				sampleStartAt = hour;
			}
			if(dpsInHour.getSummary()==null){
				continue;
			}
			Number value = this.sampleFromSummary(dpsInHour.getSummary(), sample);
			
			if(hour-sampleStartAt+1<=frequenceLen){
				//把 同一采样周期的数据放入集合
				subList.put(hour,value);				
			}else{
				//对前一个采样周期的数据进行采样
				if(subList.size()>0){
					Number aggregated = doAggregateSample(subList,sample);
					result.put(Long.toString(sampleStartAt*3600*1000), aggregated);
				}
				//开始新的采样周期：清除前一个采样周期产生的集合数据，重置采样开始时间
				subList.clear();
				subList.put(hour,value);
				sampleStartAt = hour;
			}
		}
		if(subList.size()>0){
			Number aggregated = doAggregateSample(subList,sample);
			result.put(Long.toString(sampleStartAt*3600*1000), aggregated);
		}
		return result;
	}
	
	/**
	 * 从原始数据降频采样
	 * @param dpsInHours
	 * @param sample
	 * @return
	 */
	private Map<String,Number> sampleFromDetail(List<DpsPack> dpsInHours,DownSample sample){
		int frequenceLen = changeTimeLenToSeconds(sample.getFrequence(),sample.getUnit());
		long sampleStartAt = 0;
		Map<String,Number> result = new LinkedHashMap<String,Number>();
		Map<Long,Number> subList = new LinkedHashMap<Long,Number>();
		for(DpsPack dpsInHour:dpsInHours){			
			long hour = dpsInHour.getHour();
			long second = hour*3600;
			
			Iterator<Entry<Integer,Number>> iter = dpsInHour.getDps().entrySet().iterator();
			while(iter.hasNext()){
				Entry<Integer,Number> entry = iter.next();
				int secondInHour = entry.getKey();
				long time = second + secondInHour;
				if(sampleStartAt==0){
					sampleStartAt = frequenceLen *( time/frequenceLen);
				}
				Number value = entry.getValue();
				if(time-sampleStartAt+1<=frequenceLen) 
					//把同一个采样周期的数据放入集合
					subList.put(time, value);
				else{
					//跨入新采样周期之前，先对前一个周期进行采样
					Number aggregated = doAggregateSample(subList,sample);
					result.put(Long.toString(sampleStartAt*1000), aggregated);
					//开始新的采样周期：清除前一个采样周期产生的集合数据，重置采样开始时间
					subList.clear();
					subList.put(time, value);
					sampleStartAt = frequenceLen *( time/frequenceLen);
				}
			}
			if(subList.size()>0){
				Number aggregated = doAggregateSample(subList,sample);
				result.put(Long.toString(sampleStartAt*1000), aggregated);
			}
		}
		return result;
	}
	
	class NumberComparator implements Comparator<Entry<Long,Number>>{
		public int compare(Entry<Long,Number> e1, Entry<Long,Number> e2) {
			Number number1 = e1.getValue();
			Number number2 = e2.getValue();
			if(number1 instanceof Float){
				Float f1 = (Float)number1;
				Float f2 = (Float)number2;
				return f1.compareTo(f2);
			}else if(number2 instanceof Integer){
				Integer i1 = (Integer)number1;
				Integer i2 = (Integer)number2;
				return i1.compareTo(i2);
			}else{
				throw new RuntimeException("can not compare these numbers:"
						+number1.getClass()+","+number2.getClass());
			}
		}
	}
	
	private Number avg(Map<Long,Number> values){
		Iterator<Entry<Long,Number>> iter = values.entrySet().iterator();
		Class<?> type = null;
		while(iter.hasNext()){
			Number n = iter.next().getValue();
			if(n instanceof Float ){
				type = Float.class;
				break;
			}else if (n instanceof Integer){
				type = Integer.class;
				break;
			}
		}
		if(type==null)
			throw new RuntimeException("can not calculate avg value on unknow type");
		if(type.equals(Float.class)){
			return avgFloat(values);
		}else{
			return avgInt(values);
		}
			
	}
	
	private int avgInt(Map<Long,Number> values){
		int count = 0;
		int sum = 0;
		Set<Entry<Long,Number>> set = values.entrySet();
		for(Entry<Long,Number> e:set){
			Integer v = (Integer) e.getValue();
			sum = sum+v;
			count++;
		}
		return sum/count;
	}
	
	private float avgFloat(Map<Long,Number> values){
		int count = 0;
		float sum = 0;
		Set<Entry<Long,Number>> set = values.entrySet();
		for(Entry<Long,Number> e : set){
			Float f = (Float)e.getValue();
			sum = sum+ f.floatValue();
			count++;
		}
		return sum/count;
	}
	
	private Number doAggregateSample(Map<Long,Number> values,DownSample sample){
		Arithmetic arithmetic = sample.getArithmetic();
		if(arithmetic.equals(Arithmetic.max)){
			Optional<Entry<Long,Number>> max = values.entrySet().stream().max(new NumberComparator());
			return max.get().getValue();
		}else if(arithmetic.equals(Arithmetic.min)){
			Optional<Entry<Long,Number>> min = values.entrySet().stream().min(new NumberComparator());
			return min.get().getValue();
		}else if(arithmetic.equals(Arithmetic.avg)){
			return this.avg(values);
		}else if(arithmetic.equals(Arithmetic.first)){
			Number first = null; 
			Iterator<Entry<Long,Number>> iter = values.entrySet().iterator();
			while(iter.hasNext()){
				Number c =  iter.next().getValue();
				if(c!=null){
					first = c;
					break;
				}
			}
			return first;
		}else if(arithmetic.equals(Arithmetic.last)){
			Number last = null; 
			Iterator<Entry<Long,Number>> iter = values.entrySet().iterator();
			while(iter.hasNext()){
				Number c =  iter.next().getValue();
				if(c!=null)
					last = c;
			}
			return last;
		}else {
			throw new RuntimeException("not yet support this sample Arithmetic:"+arithmetic.toString()) ;
		}
	}
	
	public static void main(String[] args){
		
	}
}
